package org.sera.ycdc.mq;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;

import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.io.FileUtils;

import com.ibm.mq.MQC;
import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;

public class YcdcMq {
	// 队列对象
	private MQQueueManager qMgr;
	public String voperate;
	public String vhostname;
	public int vport;
	public int vcount;
	public String vqmgrs;
	public String vlocalQueue;
	public boolean vheadQueueMsg = false;
	public File voutputdir;

	/**
	 * 获取队列管理器连接
	 */
	private void getConnMQmanager() {
		MQEnvironment.hostname = this.vhostname;// MQ服务器IP
		MQEnvironment.channel = "DC.SVRCONN"; // 队列管理器对应的服务器连接通道
		MQEnvironment.CCSID = 1381; // 字符编码
		MQEnvironment.port = this.vport; // 队列管理器的端口号
		try {
			qMgr = new MQQueueManager(this.vqmgrs);// 队列管理器名称
		} catch (MQException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 关闭队列管理器连接
	 */
	private void closeConnMQmanager() {
		if (qMgr != null) {
			try {
				qMgr.close();
			} catch (MQException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 发送消息
	 */
	public void sendMsg(String msgStr) {
		int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT
				| MQC.MQOO_INQUIRE;
		MQQueue queue = null;
		try {
			// 建立Q1通道的连接
			queue = qMgr.accessQueue(this.vlocalQueue, openOptions, null, null,
					null);
			MQMessage msg = new MQMessage();// 要写入队列的消息
			msg.format = MQC.MQFMT_STRING;
			msg.characterSet = 1381;
			msg.writeObject(msgStr); // 将消息写入消息对象中
			MQPutMessageOptions pmo = new MQPutMessageOptions();
			msg.expiry = -1; // 设置消息用不过期
			queue.put(msg, pmo);// 将消息放入队列
		} catch (MQException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (queue != null) {
				try {
					queue.close();
				} catch (MQException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 发送文件到mq队列
	 */
	public void sendMsgtoMQ() {
		try {
			String[] extensions = { "msg" };
			// 使用listFiles搜索
			Collection<File> files = FileUtils.listFiles(this.voutputdir,
					extensions, false);
			System.out.println("待发送的消息文件数为：" + files.size());
			System.out.println("===============================================");
			// 迭代输出
			for (Iterator<File> iterator = files.iterator(); iterator.hasNext();) {
				File file = (File) iterator.next();
				System.out.println("开始放入消息文件:" + file.getAbsolutePath());
				sendFileMsg(file);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 发送文件消息
	 * 
	 * @throws IOException
	 */
	public void sendFileMsg(File file) throws IOException {
		String msgStr = FileUtils.readFileToString(file);
		if (this.vheadQueueMsg && -1 != msgStr.indexOf("<?xml version=\"1.0\" encoding=\"UTF-8\"?>")) {
			// 处理死信队列转出消息头部的乱码
			System.out.println("此消息头部有异常信,去除后继续发送到目标队列中");
			msgStr = msgStr.substring(msgStr
					.indexOf("<?xml version=\"1.0\" encoding=\"UTF-8\"?>"));
		}
		int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT
				| MQC.MQOO_INQUIRE;
		MQQueue queue = null;
		try {
			// 建立Q1通道的连接
			queue = qMgr.accessQueue(this.vlocalQueue, openOptions, null, null,
					null);
			MQMessage msg = new MQMessage();// 要写入队列的消息
			msg.characterSet = 1208;
			msg.write(msgStr.getBytes("gb2312"));
			queue.put(msg, new MQPutMessageOptions());
			System.out.println("发送消息文件" + file.getAbsolutePath() + "完成");
			file.renameTo(new File(file.getAbsolutePath() + ".bak"));
			System.out.println("===============================================");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (queue != null) {
				try {
					queue.close();
				} catch (MQException e) {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 接收消息
	 */
	public void receiveMsg() {
		int openOptions = MQC.MQOO_INPUT_AS_Q_DEF | MQC.MQOO_OUTPUT
				| MQC.MQOO_INQUIRE;
		MQQueue queue = null;
		try {
			// 临时处理代码--------------------------------------------
			if (!this.voutputdir.exists()) {
				this.voutputdir.mkdirs();
			}
			queue = qMgr.accessQueue(this.vlocalQueue, openOptions, null, null,
					null); // LOCAL_DEADQ
			int currentDepth = queue.getCurrentDepth();
			System.out.println("队列当前的深度为:" + currentDepth);
			if (currentDepth < this.vcount) {
				this.vcount = currentDepth;
				System.out.println("队列当前的深度小于获取消息数，修改获取消息数为:" + currentDepth);
			}
			System.out
					.println("===============================================");
			for (int i = 0; i < this.vcount; i++) {
				// 将队列的里的消息读出来
				MQMessage msg = new MQMessage();// 要读的队列的消息
				MQGetMessageOptions gmo = new MQGetMessageOptions();
				queue.get(msg, gmo);
				System.out.println("开始写入消息：" + i);
				System.out.println("消息的大小为：" + msg.getDataLength());
				byte[] b = new byte[msg.getMessageLength()];
				msg.readFully(b);// 消息读到数组里面
				String data = new String(b, "GB2312");
				File tofile = File.createTempFile("mqx_" + i + "_", ".msg",
						this.voutputdir);
				FileUtils.writeStringToFile(tofile, data);
				System.out.println("写入文件地址：" + tofile.getAbsolutePath());
				System.out.println("消息" + i + "写入文件完成");
				System.out
						.println("===============================================");
				qMgr.commit();
			}
		} catch (Exception e) {
			try {
				qMgr.backout();
				System.out.println("异常,消息回滚");
			} catch (MQException e1) {
				e1.printStackTrace();
			}
			e.printStackTrace();
		} finally {
			if (queue != null) {
				try {
					queue.close();
				} catch (MQException e) {
					e.printStackTrace();
				}
			}
		}
	}

	public static void main(String[] args) {
		// Create a Parser
		CommandLineParser parser = new BasicParser();
		Options options = new Options();
		options.addOption("H", "help", false, "Print this usage information");
		String exempelStr = "Get Exempel:\njava -jar ycdcMq.jar -h 192.168.1.121 -p 1414 -m QM_11430001  -q RQ_11430001 -o get -c 10 -f  /tmp/msg\nPut Exempel:\njava -jar ycdcMq.jar -h 192.168.1.121 -p 1414 -m QM_11430001  -q LQ_11430001 -o put -f  /tmp/msg";
		options.addOption("E", "exempel", false, exempelStr);
		options.addOption("h", "host", true, "MQ host address.");
		options.addOption("p", "port", true, "MQ host port.");
		options.addOption("m", "manager", true, "the mq queue manager.");
		options.addOption("q", "queue", true, "MQ queue manager.");
		options.addOption("o", "operate", true, "Operation type: send or receive.");
		options.addOption("d", "deadqueue", false, "News from the dead letter queue, queue to the normal XML head need to remove redundant information.");
		options.addOption("c", "count", true,
				"The number of messages received, the default is 30, the transmission will ignore this parameter.");
		options.addOption("f", "filepath", true, "Message store path.");
		// Parse the program arguments
		CommandLine commandLine;

		YcdcMq mm = new YcdcMq();
		try {
			commandLine = parser.parse(options, args);
			// Set the appropriate variables based on supplied options
			if (commandLine.hasOption('H')) {
				// automatically generate the help statement
				HelpFormatter formatter = new HelpFormatter();
				formatter.printHelp("YcdcMq", options);
				System.exit(0);
			}
			if (commandLine.hasOption('E')) {
				// automatically generate the help statement
				System.out.println(exempelStr);
				System.exit(0);
			}
			if (commandLine.hasOption('h')) {
				mm.vhostname = commandLine.getOptionValue('h');
			} else {
				System.err.println("Exception: host is not allow null!");
				System.exit(1);
			}
			if (commandLine.hasOption('p')) {
				mm.vport = Integer.parseInt(commandLine.getOptionValue('p'));
			} else {
				mm.vport = 1414;
			}
			if (commandLine.hasOption('m')) {
				mm.vqmgrs = commandLine.getOptionValue('m');
			} else {
				System.err
						.println("Exception: queue manager is not allow null!");
				System.exit(1);
			}
			if (commandLine.hasOption('d')) {
				mm.vheadQueueMsg = true;
			}
			if (commandLine.hasOption('q')) {
				mm.vlocalQueue = commandLine.getOptionValue('q');
			} else {
				System.err.println("Exception: queue is not allow null!");
				System.exit(1);
			}
			if (commandLine.hasOption('o')) {
				mm.voperate = commandLine.getOptionValue('o');
			} else {
				System.err.println("Exception: operate is not allow null!");
				System.exit(1);
			}
			if (commandLine.hasOption('c')) {
				mm.vcount = Integer.parseInt(commandLine.getOptionValue('c'));
			} else {
				mm.vcount = 30;
			}
			if (commandLine.hasOption('f')) {
				mm.voutputdir = new File(commandLine.getOptionValue('f'));
			} else {
				System.err.println("Exception: filepath is not allow null!");
				System.exit(1);
			}
			mm.getConnMQmanager();
			if ("get".equals(mm.voperate)) {
				mm.receiveMsg();
			} else if ("put".equals(mm.voperate)) {
				mm.sendMsgtoMQ();
			} else {
				System.out.println("operate is error!");
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		mm.closeConnMQmanager();
	}

}
